package main

import (
	"bytes"
	"context"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/go-redis/redis/v8" // 正确的导入路径
	"net/http"
	"time"
)

const (
	broker     = "juqiao.bodyta.com"
	port       = 1883
	topic      = "subscribe/10028"
	username   = "fanmi"
	password   = "public"
	clientID   = "fanmi"
	baseURL    = "https://hxmelon.com/jq/monitor/data/api"
	redisQueue = "mqtt_queue" // Redis 队列名称
)

var (
	redisClient *redis.Client
	ctx         = context.Background()
	httpClient  = &http.Client{
		Timeout: 10 * time.Second,
		Transport: &http.Transport{
			MaxIdleConns:        10,
			MaxIdleConnsPerHost: 10,
			IdleConnTimeout:     30 * time.Second,
		},
	}
)

func main() {
	// 初始化 Redis 客户端
	redisClient = redis.NewClient(&redis.Options{
		Addr:     "localhost:6379", // Redis 地址
		Password: "",               // Redis 密码，如果没有可以为空
		DB:       0,                // 使用的 Redis 数据库编号
	})

	// 检查 Redis 连接
	if err := redisClient.Ping(ctx).Err(); err != nil {
		fmt.Printf("Redis 连接失败: %s\n", err)
		return
	}
	fmt.Println("Redis 连接成功")

	// 配置 MQTT 客户端
	opts := mqtt.NewClientOptions()
	opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
	opts.SetClientID(clientID)
	opts.SetUsername(username)
	opts.SetPassword(password)
	opts.OnConnect = func(c mqtt.Client) {
		fmt.Println("MQTT 连接成功")
		if token := c.Subscribe(topic, 1, messageHandler); token.Wait() && token.Error() != nil {
			fmt.Printf("订阅主题失败: %s\n", token.Error())
		}
	}

	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		fmt.Printf("MQTT 连接失败: %s\n", token.Error())
		return
	}

	// 启动 Redis 消息推送 Goroutine
	go processRedisQueue()

	select {}
}

func messageHandler(client mqtt.Client, msg mqtt.Message) {
	// 打印接收到的消息
	fmt.Printf("接收到的消息: %s\n", string(msg.Payload()))

	// 将消息存入 Redis 队列
	err := redisClient.LPush(ctx, redisQueue, msg.Payload()).Err()
	if err != nil {
		fmt.Printf("写入 Redis 队列失败: %s\n", err)
	} else {
		fmt.Println("消息已存入 Redis 队列")
	}
}

func processRedisQueue() {
	ticker := time.NewTicker(30 * time.Second) // 每 30 秒处理一次 Redis 队列
	defer ticker.Stop()

	for range ticker.C {
		for {
			// 从 Redis 队列中取出消息
			rawData, err := redisClient.RPop(ctx, redisQueue).Bytes()
			if err == redis.Nil {
				// 队列为空，退出当前循环
				break
			} else if err != nil {
				fmt.Printf("从 Redis 队列读取失败: %s\n", err)
				break
			}

			// 推送消息到 API
			fmt.Printf("推送数据: %s\n", string(rawData))
			pushToAPI(rawData)
		}
	}
}

func pushToAPI(data []byte) {
	req, err := http.NewRequest("POST", baseURL, bytes.NewReader(data))
	if err != nil {
		fmt.Printf("创建 HTTP 请求失败: %s\n", err)
		return
	}

	req.Header.Set("Content-Type", "application/json")
	resp, err := httpClient.Do(req)
	if err != nil {
		fmt.Printf("HTTP 请求失败: %s\n", err)
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		fmt.Printf("HTTP 响应失败: 状态码: %d\n", resp.StatusCode)
	} else {
		fmt.Println("数据推送成功")
	}
}
